package com.lagou.client;


import com.lagou.dto.RpcDto;
import com.lagou.handler.UserClientHandler;
import com.lagou.request.RpcDecoder;
import com.lagou.request.RpcEncoder;
import com.lagou.request.RpcRequest;
import com.lagou.serializer.JSONSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 消费者
 */
public class RPCConsumer {
    public static ConcurrentHashMap<String, RpcDto> rpcDtoConcurrentHashMap = new ConcurrentHashMap<>();
    private static Logger logger = LoggerFactory.getLogger(RPCConsumer.class);
    //1.创建一个线程池对象  -- 它要处理我们自定义事件
    private static ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    private static CuratorFramework client;

    private static void initCuratorFramework() throws Exception {
        //不使用fluent编程风格
        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(10, 1);

        // 使用fluent编程风格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(exponentialBackoffRetry)
                .namespace("base")  // 独立的命名空间 /base
                .build();

        client.start();
        System.out.println("会话创建了");
        RPCConsumer.client = client;
        // 创建节点
        String path = "/";
        TreeCache treeCache = new TreeCache(client, path);
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                ChildData eventData = event.getData();
                switch (event.getType()) {
                    case NODE_ADDED:
                        logger.warn(path + "节点添加" + eventData.getPath() + "\t添加数据为：" + new String(eventData.getData()));
                        add(new String(eventData.getData()));
                        break;
                    case NODE_UPDATED:
                        logger.warn(eventData.getPath() + "节点数据更新\t更新数据为：" + new String(eventData.getData()) + "\t版本为：" + eventData.getStat().getVersion());
                        break;
                    case NODE_REMOVED:
                        logger.warn(eventData.getPath() + "节点被删除");
                        remove(new String(eventData.getData()));
                        break;
                    default:
                        break;
                }
            }
        });
        treeCache.start();

        List<String> strings = client.getChildren().forPath(path);
        for (String p : strings) {
            GetDataBuilder data = client.getData();
            byte[] bytes = data.forPath("/" + p);
            if (bytes.length != 0) {
                // 数据内容
                String value = new String(bytes);
                System.out.println("获取到的节点数据内容：" + value);
                add(value);
            }
        }

    }

    public static Object createProxy(Class<?> serviceClass, final String providerParam) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    @Override
                    public Object invoke(Object o, Method method, Object[] objects) throws Exception {
                        if (RPCConsumer.client == null) {
                            initCuratorFramework();
                        }
                        if (providerParam == null) {
                            throw new RuntimeException("providerParam is requried");
                        }
                        String[] split = providerParam.split("#");
                        if (split.length < 2) {
                            throw new RuntimeException("providerParam is error");
                        }
                        String mm = split[1];

                        RpcRequest rpcRequest = new RpcRequest();
                        rpcRequest.setClassName(serviceClass.getName());
                        Method[] declaredMethods = serviceClass.getDeclaredMethods();
                        for (Method m : declaredMethods) {
                            if (m.getName().equals(mm)) {
                                rpcRequest.setMethodName(m.getName());
                            }
                        }
                        if (rpcRequest.getMethodName() == null) {
                            throw new RuntimeException("providerParam is error");
                        }

                        rpcRequest.setParameters(new String[]{(String) objects[0]});
                        rpcRequest.setParameterTypes(new Class[]{String.class});
                        rpcRequest.setRequestId(split[0]);

                        Object[] toArray = rpcDtoConcurrentHashMap.values().toArray();
                        int size = rpcDtoConcurrentHashMap.size();
                        int i = new Random().nextInt(size);
                        RpcDto rpcDto = (RpcDto) toArray[i];
                        if (rpcDto == null) {
                            return null;
                        }
                        rpcDto.getUserClientHandler().setRpcRequest(rpcRequest);
                        Object object = executorService.submit(rpcDto.getUserClientHandler()).get();
                        return object;

                    }
                });
    }


    public static void remove(String p) {
        if (rpcDtoConcurrentHashMap.containsKey(p)) {
            RpcDto rpcDto = rpcDtoConcurrentHashMap.get(p);
            ChannelFuture connect = rpcDto.getConnect();
            connect.channel().close();
            connect.cancel(true);
            rpcDtoConcurrentHashMap.remove(p);

        }

    }

    // value ip:port
    public static void add(String value) throws Exception {
        if (value == null || rpcDtoConcurrentHashMap.containsKey(value)) {
            return;
        }
        // 编写方法,初始化客户端
        //1) 初始化UserClientHandler
        UserClientHandler userClientHandler = new UserClientHandler();
        //2)创建连接池对象
        EventLoopGroup group = new NioEventLoopGroup();
        //3)创建客户端的引导对象
        Bootstrap bootstrap = new Bootstrap();
        //4)配置启动引导对象
        bootstrap.group(group)
                //设置通道为NIO
                .channel(NioSocketChannel.class)
                //设置请求协议为TCP
                .option(ChannelOption.TCP_NODELAY, true)
                //监听channel 并初始化
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //获取ChannelPipeline
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        //设置编码
                        pipeline.addLast(new RpcDecoder(String.class, new JSONSerializer()));
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        //添加自定义事件处理器
                        pipeline.addLast(userClientHandler);
                    }
                });
        String[] split = value.split(":");

        String ip = split[0];
        Integer port = Integer.valueOf(split[1]);
        ChannelFuture connect = bootstrap.connect(ip, port);
        connect.sync();

        RpcDto rpcDto = new RpcDto();
        rpcDto.setConnect(connect);
        rpcDto.setUserClientHandler(userClientHandler);
        rpcDtoConcurrentHashMap.put(value, rpcDto);
    }


}
